/*
 * Copyright 2021-2022 Open Kunlun Technology <https://www.openkunlun.io>
 */

package io.openkunlun.scaladsl.serialization

import akka.actor.{ ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider }
import com.google.protobuf.ByteString
import org.json4s.jackson.Serialization
import org.json4s.{ DefaultFormats, Serializer }

import scala.reflect.ManifestFactory

/**
 * @author ericxin.
 */
object JsonDelegateSerialization extends ExtensionId[JsonDelegateSerialization] with ExtensionIdProvider {
  override def lookup: ExtensionId[JsonDelegateSerialization] = JsonDelegateSerialization
  override def createExtension(system: ExtendedActorSystem) = new JsonDelegateSerialization(system)
}

class JsonDelegateSerialization(system: ExtendedActorSystem) extends Extension with DaprObjectSerialization {
  private val lock = new Object()
  implicit val formats = DefaultFormats

  var customerFormats: Seq[Serializer[_]] = Seq.empty

  override def serialize[T](serializable: T): Array[Byte] = {
    DelegateManifest(
      serializable.getClass.getName,
      ByteString.copyFrom(Serialization.write(serializable.asInstanceOf[AnyRef])(formats ++ customerFormats).getBytes("UTF-8"))
    ).toByteArray
  }

  /**
   *
   * @param bytes
   * @tparam T
   * @return
   */
  override def deserialize[T: Manifest](bytes: Array[Byte]): T = {
    val format = DelegateManifest.parseFrom(bytes)
    val clazz = system.dynamicAccess.getClassFor[T](format.manifest).get
    Serialization.read(new String(format.payload.toByteArray))(formats ++ customerFormats, ManifestFactory.classType(clazz))
  }

  def addFormat(append: Serializer[_]): Unit = {
    addFormats(append :: Nil)
  }

  def addFormats(appends: Seq[Serializer[_]]): Unit = {
    lock.synchronized {
      customerFormats = this.customerFormats ++ appends
    }
  }
}